/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.ycl.wx.nrd.update.worker;

import cn.ac.iie.di.ban.ycl.idgen.IDSHAGeneratorImpl;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.MD5Tools;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.MongoHelper;
import cn.ac.iie.di.ban.ycl.wx.nrd.update.util.SenderPool;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.REFieldType;
import cn.ac.iie.di.datadock.rdata.exchange.client.core.session.receive.REAbstractReceiveMessageHandler;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.REMessage;
import cn.ac.iie.di.datadock.rdata.exchange.client.v1.session.RESendSession;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.log4j.Logger;

/**
 *
 * @author zwlin
 */
public class ReceiveHandler extends REAbstractReceiveMessageHandler<REMessage> {

    static private Logger logger = Logger.getLogger(ReceiveHandler.class.getName());
    static private final HashMap<String, REFieldType> types;

    static {
        types = new HashMap<>();
        types.put("_id", REFieldType.String);
        types.put("_key", REFieldType.String);
        types.put("wx_key_fng", REFieldType.String);
        types.put("_spec", REFieldType.String);
        types.put("_ch", REFieldType.Long);
        types.put("ut", REFieldType.Long);
        types.put("ct", REFieldType.Long);
    }

    static private final String[] lookupFields = {"_id", "_key", "_spec", "_ch", "ct", "wx_key_fng", "ut"};
//    static private final String[] lookupFields = {"_id","ct","wx_key_fng","ut"};
    private boolean stopped;
    private final AtomicInteger aliveCount;

    private final MongoHelper[] mongos;
    private final Random rand;
    private final SenderPool[] pools;

    public ReceiveHandler(MongoHelper[] mongos, SenderPool[] pools) {
        this.mongos = mongos;
        this.rand = new Random(System.currentTimeMillis());
        this.pools = pools;
        stopped = false;
        aliveCount = new AtomicInteger(0);
    }

    static private IDSHAGeneratorImpl idgenerator = new IDSHAGeneratorImpl();

    static private String getId(String wx_uid) {
        try {
            return idgenerator.getId("40", "U-WX03-A", "Wechat", "Wechat", wx_uid);
        } catch (Exception ex) {
            return null;
        }
    }

    @Override
    public boolean handle(REMessage message) {
        //env
//         logger.info("haha");
        if (stopped) {
            try {
                Thread.sleep(500);
            } catch (InterruptedException ex) {
            }
            return false;
        }
        aliveCount.getAndIncrement();
        logger.info("receive record size: " + message.data.size());
        //handle
        try {
            //get idx;
            int wxIdx = -1;
            int ctIdx = -1;
            for (int i = 0; i < message.colNames.length; ++i) {
                if ("_id".equals(message.colNames[i])) {
                    wxIdx = i;
                } else if ("ct".equals(message.colNames[i])) {
                    ctIdx = i;
                } else if (wxIdx >= 0 && ctIdx >= 0) {
                    break;
                }
            }
            if (wxIdx < 0) {
                logger.error("_id column not exist!");
                return true;
            }
            if (ctIdx < 0) {
                logger.error("ct column not exist!");
                return true;
            }

            //get needed cols from message
            
            HashMap<String, HashMap<String, Object>> valueMap2keyfng = new HashMap<String, HashMap<String, Object>>();
            for (Object[] rec : message.data) {
                HashMap<String, Object> valueMap = new HashMap<>(message.data.size());
                if (rec[ctIdx] == null) {
                    continue;
                }
                //String _id = getId((String) rec[wxIdx]);
                String _id = (String) rec[wxIdx];
                String _key = _id.split("_")[2];
                String wx_key_fng = MD5Tools.getMd5(_key);
                Object ct = rec[ctIdx];
                Long ut = (long) System.currentTimeMillis() / 1000;
                valueMap.put("ct", ct);
                valueMap.put("wx_key_fng", wx_key_fng);
                valueMap.put("ut", ut);
                valueMap2keyfng.put(_id, valueMap);
//                logger.info(_id+":"+ct+"---"+ut+":"+wx_key_fng);
            }

            //update mongo
            for (MongoHelper mongo : mongos) {
//                mongo.update("_id", "ct", valueMap);
                mongo.update("_id", valueMap2keyfng);
            }

            //pick up mongo to find things.
//			MongoHelper findMongo = mongos[rand.nextInt(mongos.length)];
//            MongoHelper findMongo = mongos[1];
//            //find user records
//            List<HashMap<String, Object>> rslt = findMongo.find("_id", valueMap2keyfng.keySet(), lookupFields);
//
//            //send rslts to rmq
//            for (SenderPool pool : pools) {
//                RESendSession session = pool.BorrowSession();
//                for (HashMap<String, Object> rec : rslt) {
//                    for (Map.Entry<String, Object> entry : rec.entrySet()) {
//                        if (types.containsKey(entry.getKey()) && entry.getValue() != null) {
//                            switch (types.get(entry.getKey())) {
//                                case String:
//                                    session.setString(entry.getKey(), (String) entry.getValue());
//                                    break;
//                                case Long:
//                                    session.setLong(entry.getKey(), (long) entry.getValue());
//                                    break;
//                            }
//                        }
//                    }//end for (Map.Entry<String, Object> entry : rec.entrySet()) {
//                    while (true) {
//                        try {
//                            session.add();
//                            break;
//                        } catch (Exception e) {
//                            logger.warn("cannot add record because of " + e.getMessage() + ". try to flush first.", e);
//                            while (true) {
//                                try {
//                                    session.flush();
//                                    break;
//                                } catch (Exception ee) {
//                                    logger.warn("cannot flush records because of " + ee.getMessage() + ".", ee);
//                                }
//                            }
//                        }
//                    }//end while (true) {
//                }//end for (HashMap<String, Object> rec : rslt) {
//                while (true) {
//                    try {
//                        session.flush();
//                        break;
//                    } catch (Exception ee) {
//                        logger.warn("cannot flush records because of " + ee.getMessage() + ".", ee);
//                    }
//                }
//                logger.info("send record size: " + rslt.size());
//                if (session != null) {
//                    pool.ReturnSession(session);
//                }
//            }
        } catch (Exception ex) {
            logger.error("some thing error occur when do delayed init!", ex);
        } finally {
            aliveCount.getAndDecrement();
        }
        return true;
    }

    public void stopGracefully() {
        stopped = true;
        while (aliveCount.get() > 0) {
            try {
                Thread.sleep(200);
            } catch (InterruptedException ex) {
                logger.warn("sleep interrupted.", ex);
            }
        }
    }

}
